from kafka import KafkaProducer
from kafka.errors import KafkaError


class CrawlerJobProducer:
    def __init__(self, kafkahost, kafkaport, kafkatopic):
        self.kafkaHost = kafkahost
        self.kafkaPort = kafkaport
        self.kafkatopic = kafkatopic
        self.producer = KafkaProducer(bootstrap_servers='{kafka_host}:{kafka_port}'.format(
            kafka_host=self.kafkaHost,
            kafka_port=self.kafkaPort
            ))

    def sendmesg(self, params):
        try:
            params_message = params
            producer = self.producer
            producer.send(self.kafkatopic, params_message.encode('utf-8'))
            producer.flush()
        except KafkaError as e:
            print(e)


if __name__ == '__main__':
    crawler_producer = CrawlerJobProducer("192.168.56.10", 9092, "crawler.queue")

    for i in range(3):
        msg = "msg%d" % i
        crawler_producer.sendmesg("CECT Crawler" + msg)

    print("Produce OK!")
